package com.atguigu.flink.datastreamapi.source;

import com.atguigu.flink.pojo.WordCount;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/3/31
 *
 *  连接器： 使用flink连接外部的系统，用来读取或写入数据。
 *              从文件系统中去读，写，需要引入 FileConnector
 *              从kafka中去读，写，需要引入 KafkaConnector
 *              这些连接器都是独立的jar包，需要在pom中引入！打包时，也需要打进去！默认安装的flink环境中没有！
 *

 *  读文件，需要额外导入 FileConnector。
 *
 */
public class Demo2_FileSource
{
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //DataStreamSource<String> ds = env.readTextFile("data/words.txt");

        /*
            forRecordStreamFormat(
            final StreamFormat<T> streamFormat,:  hadoop中的输入格式。
                        普通文本文件: TextInputFormat
                        hadoop的SequnceFile:  SequenceFileFormat
             final Path... paths: 文件的路径，可以是目录，也可以是文件，可以传多个
                                读hdfs，必须引入hadoop的client依赖。
             )
         */
        FileSource<String> source = FileSource.forRecordStreamFormat(
            new TextLineInputFormat(),
            //new Path("data/")
            new Path("hdfs://hadoop102:8020/flinklogs")
        ).build();

        /*
            把数据源添加到环境中使用

            老的api:  env.addSource(SourceFunction x);
            新的api: env.fromSource(
                    Source<OUT, ?, ?> source, 构造的Source对象
                        WatermarkStrategy<OUT> timestampsAndWatermarks: 水印策略，后续再讲
                                 WatermarkStrategy.noWatermarks()： 不产生水印的水印策略
            String sourceName: 给source起个名字，自己看
            )
         */
        env.fromSource(source, WatermarkStrategy.noWatermarks(),"myfile")
           .print();



        env.execute();

    }
}
